package trouger.dmt;


import org.apache.hadoop.mapred.JobConf;

/**
 * This Runner class is used to do join-tables operations.
 * <br>The default implementation support two-table join operations.
 * Call {@link BaseMapredRunner#addKeyGroup} to specify the join key.
 * If you want to do join operation on more than two tables, you should provide your own implementation of the RecordGroupFilter interface.
 */
public class JoinTableRunner 
extends JoinableRunnerBase{
	
	public static String JOINTYPE_PARAM_NAME = "trouger.dmt.join-table.join-type";
	public static String FIRST_FORCE_UNIQUE_PARAM_NAME = "trouger.dmt.join-table.first-force-unique";
	public static String SECOND_FORCE_UNIQUE_PARAM_NAME = "trouger.dmt.join-table.second-force-unique";
	
	protected JoinType joinType = null;
	protected boolean firstSourceForceUnique = false;
	protected boolean secondSourceForceUnique = false;
	
	/**
	 * If a source table is forced to be "unique", it means that
	 * for any given key, there must be at most one record found in the table.
	 * if in fact there are more than one records found by a key,
	 * the join operation will choose the last records read to be the "unique" one.
	 * @param sourceTable index(one-based) of the source table you want to force uniqueness
	 */
	public void setForcedUniqueness(int sourceTable){
		if (sourceTable == 1){
			firstSourceForceUnique = true;
		}else if (sourceTable == 2){
			secondSourceForceUnique = true;
		}else{
			throw new IllegalArgumentException("Forced-uniqueness is specified for default two-table join operation, so the value of sourceTable parameter must be one or two");
		}
	}
	/**
	 * Set the join type of this {@link JoinTableRunner} object.
	 * For more details about join types, see http://en.wikipedia.org/wiki/Join_(SQL)
	 * @param joinType the join type. Default value is JoinType.InnerJoin
	 * @see JoinType
	 */
	public void setJoinType(JoinType joinType){
		this.joinType = joinType;
	}
	
	@Override 
	protected void configureJob(JobConf conf){
		super.configureJob(conf);
		conf.set(JoinableReducer.JOINTABLE_OR_SELFAGGREGATION_PARAM_NAME, "0");
		if (this.joinType != null){
			conf.set(JoinTableRunner.JOINTYPE_PARAM_NAME, this.joinType.toString());
		}
		if (this.firstSourceForceUnique == true){
			conf.set(JoinTableRunner.FIRST_FORCE_UNIQUE_PARAM_NAME, Boolean.toString(true));
		}
		if (this.secondSourceForceUnique == true){
			conf.set(JoinTableRunner.SECOND_FORCE_UNIQUE_PARAM_NAME, Boolean.toString(true));
		}
	}
}
